package com.dongnaoedu.rpc.redis.proxy;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.util.List;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSONObject;
import com.dongnaoedu.rpc.redis.spring.ConsumerListener;

@Component
public class RedisRpcClientHandler implements InvocationHandler {

	/** 客户端实例ID */
	private static String clientInstanceId = UUID.randomUUID().toString();

	private final Log log = LogFactory.getLog(getClass());
	@Autowired
	StringRedisTemplate stringRedisTemplate;

	@Autowired
	RedisMessageListenerContainer redisMessageListenerContainer;

	@Autowired
	ConsumerListener consumerListener;

	@SuppressWarnings("static-access")
	@Override
	public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
		// 发送数据到redis，表示你要调用某个服务了
		String msgId = UUID.randomUUID().toString();

		Class<? extends Object> clazz = proxy.getClass().getInterfaces()[0];

		final String channel = "provider@" + clazz.getName() + "@" + method.getName() + "@*";
		log.info("消费者rpc调用服务 {}. msgId {} :" + channel + ". " + msgId);
		// TODO 查询出所有的服务提供者（这里就是客户端负载均衡要实现的）
		/** 获得到channel名称。 这里是所有通道 */
		String providerChannel = stringRedisTemplate.execute(new RedisCallback<String>() {
			@Override
			public String doInRedis(RedisConnection connection) throws DataAccessException {
				@SuppressWarnings("unchecked")
				List<byte[]> result = (List<byte[]>) connection.execute("PUBSUB", "CHANNELS".getBytes(),
						channel.getBytes());
//				for (byte[] r : result) {
//					log.info("消费者rpc调用服务 msgId {}.result {} :" + channel + ". " + new String(r));
//				}
				if(result.isEmpty()) {
					return null;
				}
				int randomIndex = new Random().nextInt(result.size());
				String channelResult = new String(result.get(randomIndex));
				log.info("消费者rpc调用服务,负载均衡算法结果 msgId {}.channel {} :" + msgId + ". " + channel);
				return channelResult;
			}
		});

		String consumerChannel = "consumer@" + clazz.getName() + "@" + method.getName() + "@" + msgId;

		// publish数据到指定的生产者channel
		JSONObject message = new JSONObject();
		message.put("consumerChannel", consumerChannel);
		message.put("msgId", msgId);
		message.put("args", args);
		
		stringRedisTemplate.convertAndSend(providerChannel, message.toJSONString());
		CompletableFuture<Object> completableFuture = new CompletableFuture<>();
		consumerListener.addFuture(msgId,completableFuture);
		// TODO 接下来这个结果怎么拿到呢？
		// TODO 针对消费者，也提供一个channel。 这里每次调用都创建一个listener 也不合理，想办法留一个就行了
		redisMessageListenerContainer.addMessageListener(consumerListener, new ChannelTopic(consumerChannel));

		return completableFuture.get();
	}

}
